Source code for src.cmd.transform

import subprocess, sys, os
from contextlib import closing

from util import *
from ..util import get_cnxn, check_file
from ..configuration import db_info, config

def transform(cmd):
    args = cmd.args
    
    with closing(get_cnxn(db_info['stand']['name'],{'db_info':db_info['stand']})) as cnx:
        with closing(cnx.cursor()) as c:
            c.execute('SELECT COUNT(*) FROM Subject')
            num_subjects = c.fetchall()[0][0]
    
    print "Spawning Transform subprocesses..."
    _off_flags = args.offense_deletion_flags
    _off_flags = ' '.join('-d %s' % flag for flag in _off_flags)
    
    _cores = args.cores
    if _cores > 4: _cores = 4
    
    for segment in ['arrest', 'sentence']:
        create_log_dir(os.path.join(config.tmp_out_dir, 'transform', segment))
        
    for x in xrange(_cores):
        _position = x
        _num_subjects = num_subjects
        
        step = _num_subjects / _cores 
        start = step * _position
        
        if x == _cores - 1:
            step += (_num_subjects % _cores) + 1
            
        cmd.task_queue.put({
           'out_dir':  check_file(args.out_dir, as_dir = True),
           'meta_dir': check_file(args.metadata, as_dir = True),
           'verbose': args.verbose,
           'db_info': db_info['stand'],
           'position': _position,
           'tmp_out_dir': config.tmp_out_dir,
           #'recid_db_info': db_info['recid'],
           'start': start,
           'step': step,
           'log_dir': create_log_dir(os.path.join(config.log_dir, 'set_%i' % config.current_set, 'transform', str(_position))),
           'threads': args.thread,
           'juv_flag': args.juv_flag,
           'follow_period': int(args.follow_period),
           'off_flags': [int(off) for off in args.offense_deletion_flags],
        })
        
        
    cmd.start() #start the transform workers
    cmd.task_queue.join() #wait for the workers to finish
    cmd.flush() #flush the cmd logs